package com.atguigu.bigdata.chapter13;

import com.atguigu.bigdata.bean.UserBehavior;
import com.atguigu.bigdata.util.AtguiguUtil;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2022/9/8 9:02
 */
public class Flink02_High_Project_UV {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        
        env
            .readTextFile("input/UserBehavior.csv")
            .map(new MapFunction<String, UserBehavior>() {
                @Override
                public UserBehavior map(String line) throws Exception {
                    String[] data = line.split(",");
                    return new UserBehavior(
                        Long.valueOf(data[0]),
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2]),
                        data[3],
                        Long.parseLong(data[4]) * 1000  // 把s变成ms
                    );
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((ub, ts) -> ub.getTimestamp())
            )
            .filter(ub -> "pv".equals(ub.getBehavior()))
            // 如果用的windowAll, 你可以理解成只有一个key, 所以窗口处理函数中是可以使用键控状态的
            .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
            .process(new ProcessAllWindowFunction<UserBehavior, String, TimeWindow>() {
                @Override
                public void process(Context ctx,
                                    Iterable<UserBehavior> elements,
                                    Collector<String> out) throws Exception {
    
                    // 参数1: 确定布隆过滤器中存储的数据类型: int long string
                    // 参数2: 数据规模. 内部会根据这个参数和错误率来决定维数组的长度和hash算法的个数
                    // 参数3: 希望的最大错误率
                    BloomFilter<Long> bf = BloomFilter.create(Funnels.longFunnel(), 100 * 10, 0.01);
                    long count = 0;
                    for (UserBehavior ele : elements) {
                        if (bf.put(ele.getUserId())) { // 返回值是true表示这次存储成功
                            count++;
                        }
                    }
                    
                    out.collect(AtguiguUtil.toDatTime(ctx.window().getStart()) + " " + count);
    
                }
            })
            .print();
        
        
        env.execute();
    }
}
